/*
 * @Description:
 * @Author: Dxcr
 * @Date: 2025-01-15 09:53:10
 * @LastEditTime: 2025-01-16 11:10:47
 * @LastEditors: Dxcr
 */
mod config_loader;
mod mqtt_client;

use chrono::Local;
use config_loader::Config;
use config_loader::WellsiteInformation;
use csv::Reader;
use mqtt_client::MqttClient;
use serde::Deserialize;
use serde::Serialize;
use std::sync::atomic::{AtomicUsize, Ordering};

const MQTT_ID: &str = "rust_client_publisher";

#[tokio::main]
async fn main() {
    // 从配置文件加载配置
    let config = Config::load("config.json");
    println!("========================");
    println!(" 📂 配置文件（config.json）读取成功！");
    println!("========================");

    let mqtt_id: &str = MQTT_ID;
    let mqtt_broker = config.mqtt_broker;
    let mqtt_port = config.mqtt_port;
    let mqtt_username = config.mqtt_username;
    let mqtt_password = config.mqtt_password;

    let mqtt_publisher = MqttClient::new(
        mqtt_id,
        &mqtt_broker,
        mqtt_port,
        &mqtt_username,
        &mqtt_password,
    );

    let wellsite_information = config.wellsite_information;

    let wellsite_data = generate_wellsite_data(wellsite_information);
    mqtt_publisher
        .publish_interval("WellsiteInformation", &wellsite_data, 5)
        .await;

    let stagejob_datas = generate_stagejob_data();
    println!("===================================");
    println!(" 🛠️ 加载井场数据 (data.csv) 成功！");
    println!("===================================");
    let mut index = 0;
    // 回调函数：按顺序生成 payload
    let payload_generator = move || {
        let mut stagejob = stagejob_datas[index].clone();
        index = (index + 1) % stagejob_datas.len(); // 循环更新索引
        let now = Local::now().format("%Y-%m-%d %H:%M:%S").to_string();
        stagejob.timestamp = now;
        let csv= stagejob.to_csv();
        csv
    };
    mqtt_publisher
        .publish_interval_callback("StageJobData", payload_generator, 1)
        .await;
    //保持eventloop，否则不会发送
    mqtt_publisher
        .subscribe("never_not_used_keep_publish", |_, _| {})
        .await;
    // 防止主线程退出
    loop {
        std::thread::sleep(std::time::Duration::from_secs(60));
    }
}

// 生成 CSV 数据
fn generate_wellsite_data(wellsite_information: WellsiteInformation) -> String {
    let well_name = wellsite_information.well_name;
    let stage_number = wellsite_information.stage_number;
    let construction_date = wellsite_information.construction_date;
    let now = Local::now().format("%Y-%m-%d %H:%M:%S").to_string();
    let data = WellsiteInformation {
        well_name,
        stage_number,
        construction_date,
    };

    let csv_string = data.to_csv();
    csv_string
}

// 定义全局的 AtomicUsize，用于记录递增的序号
static SEQUENCE_COUNTER: AtomicUsize = AtomicUsize::new(1);

#[derive(Serialize, Clone)]
struct StagejobData {
    timestamp: String,              // 时间
    sequence_number: u32,           // 序号
    fluid_type: String,             // 液性
    fluid_viscosity: f64,           // 液体粘度 (cps)
    pressure: f64,                  // 施工压力 (MPa)
    displacement: f64,              // 排量 (m³/min)
    sand_concentration: f64,        // 砂浓度 (kg/m³)
    sand_ratio: f64,                // 砂比 (%)
    total_sand_volume: f64,         // 累计砂量 (m³)
    proppant_type: String,          // 支撑剂类型
}

impl StagejobData {
    /// 将 StagejobData 转换为 CSV 格式的字符串
    pub fn to_csv(&self) -> String {
        format!(
            "{},{},{},{},{},{},{},{},{},{};",
            self.timestamp,
            self.sequence_number,
            self.fluid_type,
            self.fluid_viscosity,
            self.pressure,
            self.displacement,
            self.sand_concentration,
            self.sand_ratio,
            self.total_sand_volume,
            self.proppant_type
        )
    }
}

fn generate_stagejob_data() -> Vec<StagejobData> {
    let now = Local::now().format("%Y-%m-%d %H:%M:%S").to_string();

    // 获取并递增序号
    let sequence_number = SEQUENCE_COUNTER.fetch_add(1, Ordering::SeqCst);

    let file_path = "data.csv";
    let records = read_csv_data(file_path);
    // for record in records {
    //     println!("{:?}", record);
    // }
    let stagejob_datas: Vec<StagejobData> = records
        .into_iter()
        .enumerate()
        .map(|(index, record)| StagejobData {
            timestamp: now.to_string(),
            sequence_number: sequence_number as u32,
            fluid_type: "高粘液".to_string(), // 固定值
            fluid_viscosity: 12.8,            // 固定值
            pressure: record.pressure.parse::<f64>().unwrap_or(0.0),
            displacement: record.displacement.parse::<f64>().unwrap_or(0.0),
            sand_concentration: 112.7, // 固定值
            sand_ratio: record.sand_ratio.parse::<f64>().unwrap_or(0.0),
            total_sand_volume: record.total_sand_volume.parse::<f64>().unwrap_or(0.0),
            proppant_type: "70/140目陶粒".to_string(), // 固定值
        })
        .collect();

    stagejob_datas
}

// 定义数据结构
#[derive(Debug, Deserialize)]
struct Record {
    #[serde(rename = "施工泵压")]
    pressure: String,
    #[serde(rename = "排量")]
    displacement: String,
    #[serde(rename = "砂比")]
    sand_ratio: String,
    #[serde(rename = "累计砂量")]
    total_sand_volume: String,
}

// 封装 CSV 数据读取为方法
fn read_csv_data(file_path: &str) -> Vec<Record> {
    let mut rdr = match Reader::from_path(file_path) {
        Ok(reader) => reader,
        Err(err) => {
            eprintln!("Error opening file: {}", err);
            return Vec::new(); // 返回空列表
        }
    };

    let mut records = Vec::new();

    for result in rdr.deserialize() {
        match result {
            Ok(record) => records.push(record),
            Err(err) => eprintln!("Error parsing record: {}", err),
        }
    }

    records // 返回读取到的记录
}
